package analysis_uv;

import beans.PageViewCount;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

/**
 * @author zkq
 * @date 2022/10/5 18:59
 */
public class UvWithBloom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<UserBehavior> inputStream = env
                .readTextFile("F:\\javasecode220620\\UserBehaviorAnalysis\\NetworkFlowAnalysis\\src\\main\\resources\\UserBehavior.csv")
                .map(data -> {
                    String[] splits = data.split(",");
                    return new UserBehavior(Long.parseLong(splits[0]), Long.parseLong(splits[1]), Integer.parseInt(splits[2]),
                            splits[3], Long.parseLong(splits[4]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                            @Override
                            public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                return element.getTimestamp() * 1000;
                            }
                        })
                );
        //开窗统计 一小时统计一次结果
        SingleOutputStreamOperator<PageViewCount> process = inputStream
                .filter(data -> "pv".equals(data.getBehavior()))
                .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
                .trigger(new MyTrigger())
                .process(new UvCountResultWithBloom());

        process.print();
        env.execute("uv with bloom");
    }
    public static class MyTrigger extends Trigger<UserBehavior, TimeWindow>{
        //来一个数据触发一次 清空窗口里的数据
        @Override
        public TriggerResult onElement(UserBehavior element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.FIRE_AND_PURGE;
        }

        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {

        }
    }
    public static class MyBloomFilter{
        //位图大小 尽量给2的n次幂方便计算 尽量估算好有多少条数据后扩大一些位图大小 减少碰撞概率
        private Integer cap;

        public MyBloomFilter(Integer cap) {
            this.cap = cap;
        }
        public Long hashCode(String value,Integer seed){
            Long result = 0L;
            for (int i = 0; i < value.length(); i++) {
                result = result * seed + value.charAt(i);
            }
            //防止超过位图大小 2的n次幂-1 得到 0001111111 做& 前面位都是0 &后还是0 相当于做了截取 不会超过位图大小
            return result & (cap - 1);
        }
    }
    public static class  UvCountResultWithBloom extends ProcessAllWindowFunction<UserBehavior,PageViewCount,TimeWindow>{
        private MyBloomFilter bloomFilter;
        private Jedis jedis;
        String uvCountMapName = "uvCount";

        //定义redis连接和布隆过滤器 存两个东西 存一个bitmap位图 存一个hash表
        //把userId用布隆过滤器存到redis的位图里 还需要准备一张hash表保存每个窗口的uv的count值
        @Override
        public void open(Configuration parameters) throws Exception {
            jedis = new Jedis("hadoop102", 6379);
            //假设 64MB 2的6 * 2的20 * 2的3 2的29次方
            bloomFilter = new MyBloomFilter(1<<29);
        }


        @Override
        public void process(Context context, Iterable<UserBehavior> elements, Collector<PageViewCount> out) throws Exception {
            long windowEnd = context.window().getEnd();
            String bitmapKey = String.valueOf(windowEnd);
            String uvCountKey = String.valueOf(windowEnd);

            String userId = elements.iterator().next().toString();
            Long offset = bloomFilter.hashCode(userId,61);
            //判断当前user是否存在于redis的位图中
            Boolean isExit = jedis.getbit(bitmapKey,offset);
            //不存在 就添加进去 然后uvCount + 1 也需判断是否为空 为空就初始化为1 不为空就取出结果在原结果基础上+1
            if(!isExit){
                //将位图中对应位置置为1
                jedis.setbit(bitmapKey,offset,true);
                Long uvCount = 0L;
                String hget = jedis.hget(uvCountMapName, uvCountKey);
                if(hget!=null && !"".equals(hget)){
                    uvCount = Long.valueOf(hget);
                }
                jedis.hset(uvCountMapName,uvCountKey,String.valueOf(uvCount+1));
            }

        }
        @Override
        public void close() throws Exception {
            jedis.close();
        }
    }
}
